TASK-055: emit CEG shadow comparison outputs - #178
Conversation
|
📋 Best Practices for Large Changes
✅ This PR passes the blocking limit but is larger than recommended. |
L9 Audit Harness Report
Step Results
Architecture Audit Findings
See Spec Coverage
See Next StepsAll checks passed. Safe to merge. |
|
There was a problem hiding this comment.
Pull request overview
This PR introduces an observational-only “shadow comparison” artifact for CEG, enabling deterministic JSON output that compares primary vs shadow candidate rankings without affecting production match authority.
Changes:
- Added
engine/shadow/comparison types + comparison function that emits mismatch classes (rank,score,missing,extra) and a deterministic checksum. - Added
tools/shadow_comparison.pyCLI to generate the comparison artifact from an offline JSON input. - Added unit tests plus supporting runbook + ADR for TASK-055.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| engine/shadow/compare.py | Implements ranking normalization, mismatch detection, and deterministic serialization/checksum. |
| engine/shadow/init.py | Exposes the shadow comparison API via package exports. |
| tools/shadow_comparison.py | CLI wrapper to load rankings from JSON and emit a comparison artifact JSON file. |
| tests/unit/test_shadow_comparison.py | Unit coverage for checksum determinism, mismatch classes, and CLI artifact writing. |
| docs/runbooks/CEG_SHADOW_COMPARISON.md | Operational runbook for offline emission and safety/rollback guidance. |
| docs/adr/ADR-110-ceg-shadow-comparison-outputs.md | ADR documenting the decision and constraints for shadow comparison outputs. |
| @@ -0,0 +1,136 @@ | |||
| """Deterministic primary-vs-shadow ranking comparison (observational only).""" | |||
| data = json.loads(Path(args.input).read_text()) | ||
| comparison = emit_shadow_comparison( | ||
| packet_id=str(data["packet_id"]), | ||
| primary=_load_ranked(data.get("primary") or []), | ||
| shadow=_load_ranked(data.get("shadow") or []), | ||
| ) | ||
| out = Path(args.output) | ||
| out.parent.mkdir(parents=True, exist_ok=True) | ||
| payload = comparison.to_dict() | ||
| out.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") |
| """Observational shadow comparison outputs (TASK-055). | ||
|
|
||
| Does not replace primary match authority. | ||
| """ |
| @dataclass | ||
| class ShadowComparison: | ||
| schema: str = "l9.ceg.shadow_comparison.v1" | ||
| packet_id: str = "" | ||
| observational: bool = True | ||
| replaces_primary: bool = False | ||
| primary: list[RankedCandidate] = field(default_factory=list) | ||
| shadow: list[RankedCandidate] = field(default_factory=list) | ||
| mismatches: list[Mismatch] = field(default_factory=list) | ||
| checksum: str = "" | ||
|
|
||
| def to_dict(self) -> dict[str, Any]: | ||
| body = { | ||
| "schema": self.schema, | ||
| "packet_id": self.packet_id, | ||
| "observational": self.observational, | ||
| "replaces_primary": self.replaces_primary, | ||
| "primary": [asdict(x) for x in self.primary], | ||
| "shadow": [asdict(x) for x in self.shadow], | ||
| "mismatches": [asdict(x) for x in self.mismatches], | ||
| } | ||
| blob = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False) | ||
| self.checksum = "sha256:" + hashlib.sha256(blob.encode()).hexdigest() | ||
| body["checksum"] = self.checksum | ||
| return body |




Generated under L9 controlled autonomy.
Task: TASK-055
Program: sha256:9cd1a79f948dac419913c134396e58359e4df82862bb3901bdd327684a37cb52
Contract: sha256:514d282ce3d47c1c0d2b0167d6bb77d21904b2f789e1d5996e0cd0e8d65d457f
Verification: sha256:58ac85f9a17dc29122f19c26726122af0629dd5dc340c62996e31faa2bc773ff
This PR is draft only. The controller cannot mark ready, approve, merge, tag, release, or deploy.